package blue.mqtt4j.internal.client.net;

import blue.mqtt4j.client.MqttOptions;
import blue.mqtt4j.core.handler.HeartbeatHandler;
import blue.mqtt4j.internal.client.handler.MessageHandlerFactory;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * @author Jin Zheng
 * @since 1.0 2020-09-30
 */
public class MqttInitializer extends ChannelInitializer<SocketChannel>
{
	private final NettyClient client;
	private final MqttOptions options;
	private final int idleSecs;
	private final int timeoutSecs;
	private final RetryPolicy retryPolicy;
	private final MessageHandlerFactory factory;

	public MqttInitializer(NettyClient client, MqttOptions options)
	{
		this.client = client;
		this.options = options;
		this.idleSecs = options.keepAliveSecs();
		this.timeoutSecs = options.timeoutMillis() < 1000 ? 1 : options.timeoutMillis() / 1000;
		this.retryPolicy = new DefaultRetryPolicy(options.timeoutMillis());
		this.factory = new MessageHandlerFactory(client, options);
	}

	@Override
	protected void initChannel(SocketChannel ch) throws Exception
	{
		ChannelPipeline cp = ch.pipeline();
		cp.addLast(new ReconnectHandler(client, retryPolicy));
		cp.addLast(new MqttDecoder(Short.MAX_VALUE));
		cp.addLast(MqttEncoder.INSTANCE);
		cp.addLast(new IdleStateHandler(idleSecs + timeoutSecs, idleSecs, idleSecs + timeoutSecs));
		cp.addLast(new HeartbeatHandler());
		cp.addLast(new MqttConnectHandler(client, options));
		cp.addLast(new MqttMessageHandler(factory));
	}

}
